package net.opentsdb.tsd;

import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.utils.TsdStrUtil;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.*;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Hujf
 * @title: RpcHandler
 * @date 2021-09-14 17:17
 * @description: RPC处理器  接收信息等
 */
public class RpcHandler extends IdleStateAwareChannelUpstreamHandler {

    private static final Logger LOG = LoggerFactory.getLogger(RpcHandler.class);

    /**
     * RPC plugins.  Contains the handlers we dispatch requests to.
     */
    private final RpcManager rpc_manager;

    private final TelnetRpc unknown_cmd = new Unknown();

    private static final AtomicLong telnet_rpcs_received = new AtomicLong();

    private static final AtomicLong http_plugin_rpcs_received = new AtomicLong();

    private static final AtomicLong http_rpcs_received = new AtomicLong();

    private static final AtomicInteger exceptions_caught = new AtomicInteger();
    /**
     * The TSDB to use.
     */
    private final TSDB tsdb;

    private final HashSet<String> cors_domains;

    private final String cors_headers;

    public RpcHandler(TSDB tsdb, RpcManager manager) {
        this.tsdb = tsdb;
        this.rpc_manager = manager;


        final String cors = tsdb.getConfig().getString("tsd.http.request.cors_domains");
        LOG.info("TSD is in " + tsdb.getMode() + " mode");

        if (cors == null || cors.isEmpty()) {
            cors_domains = null;
            LOG.info("CORS domain list was empty, CORS will not be enabled");
        } else {
            final String[] domains = cors.split(",");
            cors_domains = new HashSet<String>(domains.length);
            for (final String domain : domains) {
                if (domain.equals("*") && domains.length > 1) {
                    throw new IllegalArgumentException(
                            "tsd.http.request.cors_domains must be a public resource (*) or "
                                    + "a list of specific domains, you cannot mix both.");
                }
                cors_domains.add(domain.trim().toUpperCase());
                LOG.info("Loaded CORS domain (" + domain + ")");
            }
        }

        cors_headers = tsdb.getConfig().getString("tsd.http.request.cors_headers").trim();

        if ((cors_headers == null) || !cors_headers.matches("^([a-zA-Z0-9_.-]+,\\s*)*[a-zA-Z0-9_.-]+$")) {
            throw new IllegalArgumentException(
                    "tsd.http.request.cors_headers must be a list of validly-formed "
                            + "HTTP header names. No wildcards are allowed.");
        } else {
            LOG.info("Loaded CORS headers (" + cors_headers + ")");
        }
    }

    /**
     * 接收到消息之后就可以进行相关处理了  思路  对命令保存到一个hashmap中  如果有  处理 没有
     */
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        Object message = e.getMessage();
        if (message instanceof String[]) {
            handleTelnetRpc(e.getChannel(), (String[]) message);
        } else if (message instanceof HttpRequest) {
            handleHttpRpc(e.getChannel(), (HttpRequest) message);
        } else {
            LOG.info(e.getChannel() + "Unexpected message type "
                    + message.getClass() + ": " + message);
            exceptions_caught.incrementAndGet();
        }
    }

    private void handleHttpRpc(Channel channel, HttpRequest message) {

        AbstractHttpQuery abstractQuery = null;
        abstractQuery = createQueryInstance(channel, message);

        String route = message.getUri();
        final HttpRpc rpc = rpc_manager.lookupHttpRpc(route);
        HttpQuery builtinQuery = (HttpQuery) abstractQuery;
        if (rpc != null) {
            try {
                rpc.execute(tsdb, builtinQuery);

//                class GroupCB implements Callback<Object, ArrayList> {
//
//
//                    @Override
//                    public Object call(ArrayList arg) throws Exception {
//                        query.sendReply("".getBytes(StandardCharsets.UTF_8));
//                        return null;
//                    }
//                }
//                try {
//                    new GroupCB().call(new ArrayList<>());
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }

            } catch (BadRequestException e) {
                if(abstractQuery==null){

                }else{
                    abstractQuery.sendBuffer(HttpResponseStatus.BAD_REQUEST, ChannelBuffers.copiedBuffer(e.toString().getBytes(StandardCharsets.UTF_8)),"text/plain");
                }
            }
        } else {
            builtinQuery.notFound();
        }

    }

    private AbstractHttpQuery createQueryInstance(Channel channel, HttpRequest request) {
        String uri = request.getUri();
        if (TsdStrUtil.isNullOrEmpty(uri)) {
            throw new RuntimeException("Request URI is empty");
        } else if (uri.charAt(0) != '/') {
            throw new RuntimeException("Request URI doesn't start with a slash");
        } else if (rpc_manager.isHttpRpcPluginPath(uri)) {
            http_plugin_rpcs_received.incrementAndGet();
            return new HttpRpcPluginQuery(tsdb, request, channel);
        } else {
            http_rpcs_received.incrementAndGet();
            HttpQuery builtinQuery = new HttpQuery(tsdb, request, channel);
            return builtinQuery;
        }
    }

    private void handleTelnetRpc(Channel channel, String[] command) {
        TelnetRpc rpc = rpc_manager.lookupTelnetRpc(command[0]);
        if (rpc == null) {
            rpc = unknown_cmd;
        }

        telnet_rpcs_received.incrementAndGet();
        rpc.execute(tsdb, channel, command);
    }


    @Override
    public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
        if (e instanceof ChannelStateEvent) {
            LOG.info(e.toString());
        }
        super.handleUpstream(ctx, e);
    }

    /**
     * For unknown commands.
     */
    private static final class Unknown implements TelnetRpc {
        @Override
        public Deferred<Object> execute(final TSDB tsdb, final Channel chan,
                                        final String[] cmd) {
            LOG.info(chan + "unknown command : " + Arrays.toString(cmd));
            chan.write("unknown command: " + cmd[0] + ".  Try `help'.\n");
            return Deferred.fromResult(null);
        }
    }
}
